package com.hs.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.UUID;

public class KafkaTopicWatcher {

    /**
     * java -jar kafka-topic-watcher-1.0.0.jar --brokers 10.125.2.6:9094,10.125.2.6:9095,10.125.2.6:9096 --topic topic-integration-projectprogress --timeout 1200 --from-beginning false
     */
    public static void main(String[] args) {
        Args a = Args.parse(args);

        String brokers = a.brokers;
        String topic = a.topic;
        long timeoutSec = a.timeoutSec;
        boolean fromBeginning = a.fromBeginning;

        if (isBlank(brokers) || isBlank(topic)) {
            System.err.println("缺少必要参数。示例：");
            System.err.println("java -jar kafka-topic-watcher-1.0.0.jar --brokers 127.0.0.1:9092 --topic my-topic --timeout 120 --from-beginning false");
            System.exit(1);
        }

        String randomGroupId = "kafka-checker-" + UUID.randomUUID();
        String clientId = "kafka-checker-client-" + UUID.randomUUID().toString().substring(0, 8);

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, randomGroupId);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, fromBeginning ? "earliest" : "latest");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");

        // 如果需要鉴权或 SSL，请按需配置：
        // props.put("security.protocol", "SASL_PLAINTEXT");
        // props.put("sasl.mechanism", "PLAIN");
        // props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xxx\" password=\"yyy\";");
        // props.put("ssl.truststore.location", "your_truststore.jks");
        // props.put("ssl.truststore.password", "your_password");

        System.out.printf("启动消费者：brokers=%s, topic=%s, groupId=%s, fromBeginning=%s, timeout=%ds%n",   brokers, topic, randomGroupId, fromBeginning, timeoutSec);

        long start = System.currentTimeMillis();
        long timeoutMs = timeoutSec * 1000;
        boolean gotMessage = false;

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList(topic));
            while (System.currentTimeMillis() - start < timeoutMs) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if (!records.isEmpty()) {
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("收到消息: partition=%d, offset=%d, key=%s, value=%s, ts=%d%n",
                                record.partition(), record.offset(), record.key(), record.value(), record.timestamp());
                    }
                    gotMessage = true;
                    break; // 收到第一批消息即退出，如需持续监听可去掉这行
                }
            }
        } catch (Exception e) {
            System.err.println("消费异常: " + e.getMessage());
            e.printStackTrace();
            System.exit(1);
        }

        if (gotMessage) {
            System.out.println("结果：检测到新消息进入该 topic。");
            System.exit(0);
        } else {
            System.out.println("结果：在超时时间内未检测到消息进入该 topic。");
            System.exit(2);
        }
    }

    private static boolean isBlank(String s) {
        return s == null || s.trim().isEmpty();
    }

    // 简易参数解析
    static class Args {
        String brokers;
        String topic;
        long timeoutSec = 120;
        boolean fromBeginning = false;

        static Args parse(String[] args) {
            Args a = new Args();
            for (int i = 0; i < args.length; i++) {
                switch (args[i]) {
                    case "--brokers":
                        if (i + 1 < args.length) a.brokers = args[++i];
                        break;
                    case "--topic":
                        if (i + 1 < args.length) a.topic = args[++i];
                        break;
                    case "--timeout":
                        if (i + 1 < args.length) a.timeoutSec = Long.parseLong(args[++i]);
                        break;
                    case "--from-beginning":
                        if (i + 1 < args.length) a.fromBeginning = Boolean.parseBoolean(args[++i]);
                        break;
                    default:
                        // 忽略未知参数
                }
            }
            return a;
        }
    }
}
